package com.niit.DWD

import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.log4j.{Level, Logger}
import java.text.SimpleDateFormat
import java.util.Date
import org.apache.hadoop.fs.{FileSystem, Path}

object DwdUserPay {
  def main(args: Array[String]): Unit = {
    // 设置日志级别
    Logger.getLogger("org").setLevel(Level.WARN)
    Logger.getLogger("akka").setLevel(Level.WARN)
    val logger = Logger.getLogger(getClass.getName)

    // 参数处理：获取日期
    val dt = if (args.length > 0) args(0) else {
      val sdf = new SimpleDateFormat("yyyy-MM-dd")
      sdf.format(new Date())
    }

    // HDFS配置
    val hdfsUri = "hdfs://192.168.10.130:9000"
    val tableLocation = s"$hdfsUri/training/hive/warehouse/commerces/dwd/dwd_user_pay"

    // 初始化SparkSession
    val spark = SparkSession.builder()
      .appName("ODS to DWD: dwd_user_pay")
      .master("local[*]")
      .config("hive.metastore.uris", "thrift://192.168.10.130:9083")
      .config("spark.hadoop.fs.defaultFS", hdfsUri) // 确保使用HDFS
      .enableHiveSupport()
      .getOrCreate()

    try {
      logger.info(s"开始处理日期: $dt")
      spark.sql("USE commerces")

      // 检查表路径是否存在，不存在则创建
      val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
      val tablePath = new Path(tableLocation)

      if (!fs.exists(tablePath)) {
        logger.info(s"创建表路径: $tableLocation")
        fs.mkdirs(tablePath)
        fs.setPermission(tablePath, new org.apache.hadoop.fs.permission.FsPermission("775"))
      }

      // 创建表（确保表存在）
      logger.info("创建外部表 dwd_user_pay")
      spark.sql(
        s"""
           |CREATE EXTERNAL TABLE IF NOT EXISTS dwd_user_pay (
           |  user_id BIGINT COMMENT '用户ID',
           |  session_id STRING COMMENT '会话ID',
           |  page_id BIGINT COMMENT '页面ID',
           |  action_time STRING COMMENT '动作时间',
           |  city_id INT COMMENT '城市ID',
           |  pay_category_ids STRING COMMENT '支付的商品类别ID列表',
           |  pay_product_ids STRING COMMENT '支付的商品ID列表'
           |)
           |STORED AS PARQUET
           |LOCATION '$tableLocation'
        """.stripMargin)

      // 验证表结构是否匹配
      val tableSchema = spark.table("dwd_user_pay").schema
      logger.info("目标表结构: " + tableSchema.treeString)

      // 读取ODS数据，筛选有支付行为的数据
      logger.info(s"读取 $dt 的ODS数据...")
      val odsData = spark.sql(
        s"""
           |SELECT
           |  user_id,
           |  session_id,
           |  page_id,
           |  action_time,
           |  CAST(city_id AS INT) AS city_id,
           |  pay_category_ids,
           |  pay_product_ids
           |FROM ods_user_visit_action
           |WHERE date = '$dt'
           |  AND pay_product_ids IS NOT NULL
           |  AND length(pay_product_ids) > 0
        """.stripMargin)

      logger.info(s"读取到 ${odsData.count()} 条支付数据")

      // 验证源数据结构
      logger.info("源数据结构: " + odsData.schema.treeString)

      // 验证源数据与目标表结构是否匹配
      if (tableSchema.fieldNames.sorted.deep != odsData.schema.fieldNames.sorted.deep) {
        logger.error("源数据与目标表结构不匹配")
        logger.error("源字段: " + odsData.schema.fieldNames.sorted.mkString(", "))
        logger.error("目标字段: " + tableSchema.fieldNames.sorted.mkString(", "))
        throw new Exception("源数据与目标表结构不匹配")
      }

      // 写入DWD表
      logger.info("开始写入数据到 dwd_user_pay")
      odsData.write
        .mode(SaveMode.Append)
        .insertInto("dwd_user_pay")

      logger.info(s"数据写入成功！日期: $dt")

    } catch {
      case e: Exception =>
        logger.error(s"处理失败: ${e.getMessage}", e)
        e.printStackTrace()
        System.exit(1)
    } finally {
      spark.stop()
      logger.info("SparkSession已关闭")
    }
  }
}